package com.surfilter.massdata.spark.task;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.regex.Pattern;

import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import scala.Tuple2;

import com.act.sparkanalyz.log.SysLog;
import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.surfilter.massdata.spark.bean.Dns_tmp;
import com.surfilter.massdata.spark.bean.Domain;
import com.surfilter.massdata.spark.bean.ExDomain;
import com.surfilter.massdata.spark.bean.ExDomainRegex;
import com.surfilter.massdata.spark.util.BaseRedisDao;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.ConfigUtil;
import com.surfilter.massdata.spark.util.StringUtil;
import com.surfilter.massdata.spark.util.URLUtil;

public class EtlTaskNew implements ISparkTask {
	String EX_DOMAIN_REDIS_KEY = "EXDOMAIN";
	transient BaseRedisDao redisDao;
	Map<String, ExDomainRegex> exDomainMap;
	final TreeMap<Long, Long> ipMap;
	String[] DIps;
	String[] Cips;
	String URL_PREFIX_HTTP = "http://";
	String DOMIAIN_REGX = "^([a-zA-Z0-9]([a-zA-Z0-9\\-]{0,61}[a-zA-Z0-9])?\\.)+[a-zA-Z]{2,6}$";
	Pattern domainPattern;
	public EtlTaskNew() {
		redisDao = new BaseRedisDao();
		ipMap = CommonUtils.getIpExtensionMap();
	}

	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames,
			Map<String, String> commandMap) {
		List<OutQueueEntity> list = new ArrayList<OutQueueEntity>();
		this.DIps = ConfigUtil.getArray("surfilter.114.ips");
		this.Cips = ConfigUtil.getArray("surfilter.filter.cips");
		this.domainPattern = Pattern.compile(DOMIAIN_REGX);
		try {
			// 读取原始数据
			DataFrame dns_raw = dataFrames.get("d_dns_raw");
			DataFrame dnsRes = null;
			DataFrame domainRes = null;
			JavaSparkContext jsc = JavaSparkContext.fromSparkContext(dns_raw
					.sqlContext().sparkContext());
			// 发现泛域名
			JavaPairRDD<String, String> exd1 = genExdomainpre(dns_raw);
			genExdomain_new(exd1);
			// 加上本次发现的泛域名，处理dns数据
			exDomainMap = redisDao.getAllExDomainMap(EX_DOMAIN_REDIS_KEY);
			final Broadcast<Map<String, ExDomainRegex>> broadBswdMap = jsc
					.broadcast(exDomainMap);
			dnsRes = genDns_new(dns_raw, broadBswdMap.getValue());
			//dnsRes.show();
			// // 生成Domain数据
			domainRes = genDomainDataFormat_new(dnsRes, ipMap);
			//domainRes.show();
			// 输出dns数据
			list.add(new OutQueueEntity("dns", dnsRes));
			list.add(new OutQueueEntity("domain", domainRes));
		} catch (Exception e) {
			e.printStackTrace();
		}
		return list;
	}

	public DataFrame genDomainDataFormat_new(DataFrame dns, TreeMap<Long, Long> map) {
		// DIP省外IP过滤
		JavaRDD<Row> dns_filter;
		if (map != null && map.size() > 0) {
			dns_filter = dns.javaRDD().filter(new filterProvinceDIP_new(map));
		} else {
			dns_filter = dns.javaRDD();
		}
		// Dns 数据 去掉列 CIP
		// <domain dip cip dns_visit_count findtime> 转成 <domain dip :
		// dns_visit_count findtime>
		JavaPairRDD<String, String> domain_1 = dns_filter
				.mapToPair(new PairFunction<Row, String, String>() {

					@Override
					public Tuple2<String, String> call(Row line)
							throws Exception {
						String domain = line.getAs("domain").toString();
						String dip = line.getAs("dip").toString();
						String visit_count = line.getAs("dns_visit_count")
								.toString();
						String findtime = line.getAs("findtime").toString();
						String key = domain + "\t" + dip;
						String value = visit_count + "\t" + findtime;
						return new Tuple2(key, value);
					}
				});
		// printJavaPairRDD_S_I(domain_1);

		// dns_visit_count加和 <domain dip : dns_visit_count findtime>
		JavaPairRDD<String, String> domain_2 = domain_1
				.reduceByKey(new org.apache.spark.api.java.function.Function2<String, String, String>() {

					@Override
					public String call(String v1, String v2) throws Exception {
						String str1[] = StringUtils.split(v1, "\t");
						String str2[] = StringUtils.split(v2, "\t");
						int num1 = Integer.valueOf(str1[0]);
						int num2 = Integer.valueOf(str2[0]);
						int num = num1 + num2;
						long time1 = Long.valueOf(str1[1]);
						long time2 = Long.valueOf(str2[1]);
						// 取最早发现时间
						long time = (time1 < time2 ? time1 : time2);
						String value = num + "\t" + time;
						return value;
					}
				});
		// printJavaPairRDD_S_I(domain_2);

		// key:DOMAIN value:DIP1 VISITSCOUNT1 FINDTIME
		JavaRDD<Tuple2<String, String>> domain_3 = domain_2
				.map(new Function<Tuple2<String, String>, Tuple2<String, String>>() {
					@Override
					public Tuple2<String, String> call(Tuple2<String, String> v1)
							throws Exception {
						String[] key = StringUtils.split(v1._1(), "\t");
						String domain = key[0];
						String dip = key[1];
						String value = v1._2();
						return new Tuple2<String, String>(domain, dip + "\t"
								+ value);
					}
				});
		// printJavaRDD(domain_3);

		// key:DOMAIN value:DIP1 VISITSCOUNT1 FINDTIME
		JavaPairRDD<String, String> domain_pair = domain_3
				.mapToPair(new PairFunction<Tuple2<String, String>, String, String>() {

					@Override
					public Tuple2<String, String> call(
							Tuple2<String, String> line) throws Exception {
						return new Tuple2<String, String>(line._1(), line._2());
					}

				});
		// printJavaPairRDD(domain_pair);

		// <domain :dip dns_visit_count findtime> ==> <domaim : dip1|dip2
		// dns_visit_count1|dns_visit_count2 findtime>
		JavaPairRDD<String, String> domain_mult_ip = domain_pair
				.reduceByKey(new Function2<String, String, String>() {

					@Override
					public String call(String v1, String v2) throws Exception {
						// :DIP1 VISITSCOUNT1 FINDTIME
						String[] value_1 = StringUtils.split(v1, "\t");
						String[] value_2 = StringUtils.split(v2, "\t");
						long time1 = Long.valueOf(value_1[2]);
						long time2 = Long.valueOf(value_2[2]);
						// 取最早发现时间
						long time = (time1 < time2 ? time1 : time2);
						String value = value_1[0] + "|" + value_2[0] + "\t"
								+ value_1[1] + "|" + value_2[1] + "\t" + time;
						return value;
					}

				});
		// printJavaPairRDD(domain_mult_ip);

		JavaRDD<Domain> rd = domain_mult_ip
				.mapPartitions(new flatMapDomain_new());
		DataFrame domain_DF = dns.sqlContext()
				.createDataFrame(rd, Domain.class);
		return domain_DF;
	}

	private JavaPairRDD<String, String> genExdomainpre(DataFrame dns_raw) {

		// domain|cname|目的IP|源IP|IP类型|访问次数|发现时间
		// 输入：0-6.ddnsgeek.com|0-6.ddnsgeek.com|253.232.207.183|100.60.64.223|1|3|2017-05-02
		// 输出：key<目的IP topdomain> 1
		// 过滤 114Ip 、省内IP过滤、域名规则匹配
		JavaPairRDD<String, Integer> dns_filter = dns_raw.toJavaRDD()
				.filter(new Function<Row, Boolean>() {

					@Override
					public Boolean call(Row line) throws Exception {
						if (line.size() < 5) {
							return false;
						}
						String domain = line.getAs("domain");
						String dip = line.getAs("dip");
						String cip = line.getAs("cip");
						String filterDIps = "";
						for (String ip : DIps) {
							filterDIps = filterDIps + ip + ",";
						}
						// 目的IP 114IP过滤
						if (filterDIps.contains(dip)) {
							return false;
						}
						if (StringUtils.isBlank(domain)
								|| StringUtils.isBlank(dip)
								|| StringUtils.isBlank(cip)) {
							return false;
						}
						// 域名合法性校验
						if (!domainPattern.matcher(domain).matches()) {
							return false;
						}

						String topDomain = URLUtil
								.getDomainName(URL_PREFIX_HTTP + domain);
						if (topDomain == null) {
							return false;
						}
						return true;
					}

				}).mapToPair(new PairFunction<Row, String, Integer>() {

					@Override
					public Tuple2<String, Integer> call(Row t) throws Exception {
						String domain = t.getAs("domain");
						String dip = t.getAs("dip");
						return new Tuple2<String, Integer>(domain + "\t" + dip,
								1);
					}

				}).reduceByKey(new Function2<Integer, Integer, Integer>() {
					// domain IP v1
					@Override
					public Integer call(Integer v1, Integer v2)
							throws Exception {
						// TODO Auto-generated method stub
						return v1 + v2;
					}

				});

		JavaPairRDD<String, String> exd1 = dns_filter
				.mapToPair(new PairFunction<Tuple2<String, Integer>, String, String>() {

					@Override
					public Tuple2<String, String> call(Tuple2<String, Integer> t)
							throws Exception {
						// key:domain dip
						String[] key = t._1.split("\t");
						String topDomain = URLUtil
								.getDomainName(URL_PREFIX_HTTP + key[0]);
						topDomain = topDomain.toLowerCase();
						return new Tuple2<String, String>(key[1] + "\t"
								+ topDomain, "1");
					}

				});
		return exd1;
	}

	private void genExdomain_new(JavaPairRDD<String, String> exd1) {

		try {
			SysLog.info("*****************genExdomain************************");

			Map<String, Object> exd_count = exd1.countByKey();
			// System.out.println("exd_count:"+exd_count.toString());
			Iterator it = exd_count.keySet().iterator();
			int judgeNum = Integer.valueOf(ConfigUtil.get(
					"surfilter.massdata.exdomain.judge.num", "200"));
			System.out.println("judgeNum:" + judgeNum);
			Map<String, Object> exdmap = new HashMap<String, Object>();
			// 过滤掉小于阀值的key
			while (it.hasNext()) {
				String key = it.next().toString();
				int num = Integer.valueOf(exd_count.get(key).toString());
				if (num > judgeNum) {
					exdmap.put(key, exd_count.get(key).toString());
				}
			}

			// 从数据集里过滤掉小于阀值的泛域名
			JavaPairRDD<String, String> exd2 = exd1.filter(new exdFilter_new(
					exdmap));
			// key:<dip1> value:<domaim> 转为 key:<domaim> value:<dip1>
			JavaPairRDD<String, String> exd3 = exd2
					.mapToPair(new transDomainKeyMapToPair_new());
			// key:<domaim> value:<dip1,dip2>
			JavaPairRDD<String, String> exd4 = exd3
					.reduceByKey(new ipMerge_new());
			// 数据写入redis
			int maxLimit = Integer.valueOf(ConfigUtil.get(
					"surfilter.massdata.exdomain.max.limit", "10000"));
			exd4.foreachPartition(new FlatMapExDomain_new(maxLimit));
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private DataFrame genDns_new(DataFrame dns_raw,
			Map<String, ExDomainRegex> exDomainMap) {
		DataFrame dnsRes = null;
		try {
			JavaRDD<Row> dns_filter = dns_raw.toJavaRDD().filter(
					new filterIp_dns(DIps, Cips,domainPattern));
			// map key:domain + "\t" + dip + "\t" + cip
			// value: visit_count+"\t"+ 时间戳
			// 泛域名替换
			JavaPairRDD<String, String> dnsinit = dns_filter
					.mapToPair(new PairIP_new(exDomainMap, domainPattern));
			
			// reduce 计算 visit_count <value: 1+"\t"+ 时间戳>
			JavaPairRDD<String, String> dns_ret = dnsinit
					.reduceByKey(new org.apache.spark.api.java.function.Function2<String, String, String>() {
						public String call(String v1, String v2)
								throws Exception {
							String str1[] = StringUtils.split(v1, "\t");
							String str2[] = StringUtils.split(v2, "\t");
							int num1 = Integer.valueOf(str1[0]);
							int num2 = Integer.valueOf(str2[0]);
							int num = num1 + num2;
							long time1 = Long.valueOf(str1[1]);
							long time2 = Long.valueOf(str2[1]);
							// 取最早发现时间
							long time = (time1 < time2 ? time1 : time2);
							String value = num + "\t" + time;
							return value;
						}
					});
			JavaRDD<Dns_tmp> rd = dns_ret.mapPartitions(new flatMapDNS_new());
			dnsRes = dns_raw.sqlContext().createDataFrame(rd, Dns_tmp.class);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return dnsRes;
	}
}

class exdFilter_new implements Function<Tuple2<String, String>, Boolean> {
	Map<String, Object> exd_count = null;

	public exdFilter_new(Map<String, Object> exd_count) {
		this.exd_count = exd_count;
	}

	@Override
	public Boolean call(Tuple2<String, String> v1) throws Exception {
		String key = v1._1();
		if (exd_count.containsKey(key)) {
			return true;
		} else {
			return false;
		}
	}

}

class transDomainKeyMapToPair_new implements
		PairFunction<Tuple2<String, String>, String, String> {

	@Override
	public Tuple2<String, String> call(Tuple2<String, String> t)
			throws Exception {
		String[] key = StringUtils.split(t._1(), "\t");
		String ip = key[0];
		String domain = key[1];
		return new Tuple2<String, String>(domain, ip);
	}

}

class ipMerge_new implements Function2<String, String, String> {

	@Override
	public String call(String v1, String v2) throws Exception {
		if (v1.contains(v2)) {
			return v1;
		} else if (v2.contains(v1)) {
			return v2;
		} else {
			return v1 + "," + v2;
		}

	}

}

/**
 * 原始dns数据过滤：域名规则校验、Dip过滤、Cip过滤、顶级域名过滤
 */
class filterIp_dns implements Function<Row, Boolean> {
	String[] DIps;
	String[] Cips;
	Pattern domainPattern;
	filterIp_dns(String[] DIps, String[] Cips,Pattern domainPattern) {
		this.DIps = DIps;
		this.Cips = Cips;
		this.domainPattern=domainPattern;
	}

	@Override
	public Boolean call(Row line) throws Exception {

		if (line.size() < 5) {
			return false;
		}
		String URL_PREFIX_HTTP = "http://";
		String domain = line.getAs("domain").toString();
		String dip = line.getAs("dip").toString();
		String cip = line.getAs("cip").toString();

		String filterDIps = "";
		String filterCips = "";
		for (String ip : DIps) {
			filterDIps = filterDIps + ip + ",";
		}
		for (String ip : Cips) {
			filterCips = filterCips + ip + ",";
		}

		if (StringUtils.isBlank("domain") || StringUtils.isBlank("dip")
				|| StringUtils.isBlank("cip")) {
			return false;
		}
		// 域名合法性校验
		if (!domainPattern.matcher(domain).matches()) {
			return false;
		}
		// 目的IP 114IP过滤
		if (filterDIps.contains(dip)) {
			return false;
		}
		// 源IP过滤
		if (filterCips.contains(cip)) {
			return false;
		}

		String topDomain = URLUtil.getDomainName(URL_PREFIX_HTTP + domain);
		if (topDomain == null) {
			return false;
		}
		return true;
	}
}

class PairIP_new implements PairFunction<Row, String, String> {

	String URL_PREFIX_HTTP = "http://";
	Pattern domainPattern ;
	int flag = 0;
	Map<String, ExDomainRegex> exDomainMap;

	public PairIP_new(Map<String, ExDomainRegex> exDomainMap, Pattern domainPattern) {
		this.exDomainMap = exDomainMap;
		this.domainPattern = domainPattern;
	}

	@Override
	public Tuple2<String, String> call(Row line) throws Exception {
		String domain = line.getAs("domain").toString();
		String dip = line.getAs("dip").toString();
		String cip = line.getAs("cip").toString();
		String dns_visit_count = line.getAs("dns_visit_count").toString();
		String findtime = line.getAs("findtime").toString();
		// guoji.news.baidu.com
		String topDomain = URLUtil.getDomainName(URL_PREFIX_HTTP + domain);
		if (topDomain == null) {
			return null;
		}
		topDomain = topDomain.toLowerCase();
		String exDomain = domain;
		// 泛域名处理
		try {
			// 缺陷【smcs_all_v1.0.0.0】
			domain = exdomainDeal(topDomain, dip, exDomainMap);
		} catch (Exception e) {
			domain = null;
		}

		if (domain == null) {
			domain = exDomain;
			exDomain = null;
		}
		String key = "";
		key = domain + "\t" + dip + "\t" + cip;
		String value = "";
		if (this.flag == 0) {
			// findtime 转成时间戳 2017-05-02 06:00:00
			SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");// 小写的mm表示的是分钟
			java.util.Date date = sdf.parse(findtime);
			long time = date.getTime() / 1000;
			value = dns_visit_count + "\t" + time;
		} else {
			value = dns_visit_count + "\t" + findtime;
		}

		return new Tuple2(key, value);
	}

	/**
	 * 过滤掉泛域名-IP,将匹配到的泛域名-IP替换成对应的域名-IP
	 * 
	 * @param domain
	 * @param dips
	 * @return
	 */

	private String exdomainDeal(String topDomain, String dips,
			Map<String, ExDomainRegex> exDomainMap) {
		String EX_DOMAIN_DEFAULT_IP = "0.0.0.0";
		if (StringUtils.isBlank(topDomain) || StringUtils.isBlank(dips)) {
			return null;
		}
		List<String> ipList = new ArrayList<String>();
		ipList.addAll(Arrays.asList(StringUtils.split(dips, "\\|")));
		String exDomain = null;
		if (exDomainMap.containsKey(topDomain)) {
			ExDomainRegex exDomainRegex = exDomainMap.get(topDomain);
			Set<String> exIps = exDomainRegex.getIpSet();
			if (exIps.contains(EX_DOMAIN_DEFAULT_IP)) {
				exDomain = exDomainRegex.getNewName();
				return exDomain;
			} else if (intersectionSet(ipList, exIps)) {
				exDomain = exDomainRegex.getNewName();
				return exDomain;
			}
		}
		return exDomain;
	}

	private boolean intersectionSet(List<String> list, Set<String> setB) {
		list.retainAll(setB);
		return list.size() > 0;
	}
}

class FlatMapExDomain_new implements
		VoidFunction<Iterator<Tuple2<String, String>>> {
	int maxLimit = 10000;

	FlatMapExDomain_new(int maxLimit) {
		this.maxLimit = maxLimit;
	}

	@Override
	public void call(Iterator<Tuple2<String, String>> t) throws Exception {
		String EX_DOMAIN_REDIS_KEY = "EXDOMAIN";
		String EX_DOMAIN_DEFAULT_IP = "0.0.0.0";
		BaseRedisDao redisDao = new BaseRedisDao();
		ObjectMapper mapper = new ObjectMapper();
		ObjectWriter writer = mapper.writer();

		while (t.hasNext()) {
			// key:<domaim> value:<dip1,dip2>
			Tuple2<String, String> tt = t.next();
			String domain = tt._1();
			String value = tt._2();

			ExDomain exDomain = new ExDomain();
			// 当前数据找到的域名、ip
			String[] value_list = StringUtils.split(value, ",");
			// redis 里存的域名、ip
			String redis_value = redisDao.hgetExDomain(EX_DOMAIN_REDIS_KEY,
					domain);
			Set ipSet = new HashSet();
			String ips_new = "";
			if (redis_value != null) {
				ExDomain redis_domain = mapper.readValue(
						StringUtil.toLowerJson(redis_value), ExDomain.class);
				String redis_ips = redis_domain.getIps();
				String[] ip_list = StringUtils.split(redis_ips, ",");
				for (String i : ip_list) {
					ipSet.add(i.trim());
				}
			}
			if (ipSet.contains(EX_DOMAIN_DEFAULT_IP)) {

			} else {
				for (String i : value_list) {
					ipSet.add(i.trim());
				}

				if (ipSet.size() > maxLimit) {
					ips_new = EX_DOMAIN_DEFAULT_IP;
				} else {
					Iterator iterator = ipSet.iterator();
					while (iterator.hasNext()) {
						ips_new += iterator.next();
						ips_new += ",";
					}
					ips_new = ips_new.substring(0, ips_new.length() - 1);
				}

				exDomain.setDomain("www." + domain);
				exDomain.setIps(ips_new);
				exDomain.setSuffix(domain);
				exDomain.setFindtime(new SimpleDateFormat("yyyyMMdd")
						.format(new Date()));
				System.out.println("redis:"
						+ StringUtil.toUpperJson(writer
								.writeValueAsString(exDomain)));
				redisDao.hsetExDomain(EX_DOMAIN_REDIS_KEY, domain, StringUtil
						.toUpperJson(writer.writeValueAsString(exDomain)));

			}
		}
	}
}

class filterProvinceDIP_new implements Function<Row, Boolean> {
	private TreeMap<Long, Long> ipMap;

	filterProvinceDIP_new(TreeMap<Long, Long> map) {
		this.ipMap = map;
	}

	@Override
	public Boolean call(Row v1) throws Exception {
		String dip = v1.getAs("dip");
		boolean flag = CommonUtils.isExists(ipMap, dip);
		return flag;
	}
}

class flatMapDomain_new implements
		FlatMapFunction<Iterator<Tuple2<String, String>>, Domain> {

	@Override
	public Iterable<Domain> call(Iterator<Tuple2<String, String>> t)
			throws Exception {

		List<Domain> list = new ArrayList<Domain>();
		// <domaim : dip1|dip2 dns_visit_count1|dns_visit_count2 findtime>
		while (t.hasNext()) {
			Tuple2<String, String> tt = t.next();
			String key = tt._1();
			String[] value = StringUtils.split(tt._2(), "\t");
			Domain e = new Domain();
			e.setDomain(key);
			e.setDip(value[0]);
			e.setDns_visit_count(value[1]);
			e.setFindtime(Long.valueOf(value[2]));

			list.add(e);
		}
		return list;

	}

}

class flatMapDNS_new implements
		FlatMapFunction<Iterator<Tuple2<String, String>>, Dns_tmp> {

	@Override
	public Iterable<Dns_tmp> call(Iterator<Tuple2<String, String>> t)
			throws Exception {
		List<Dns_tmp> list = new ArrayList<Dns_tmp>();
		// map key:domain + "\t" + dip + "\t" + cip
		// value: 1+"\t"+ 时间戳
		while (t.hasNext()) {
			Tuple2<String, String> tt = t.next();
			String[] key = StringUtils.split(tt._1(), "\t");
			String[] value = StringUtils.split(tt._2(), "\t");
			Dns_tmp e = new Dns_tmp();
			e.setDomain(key[0]);
			e.setDip(key[1]);
			e.setCip(key[2]);
			e.setDns_visit_count(Long.valueOf(value[0]));
			e.setFindtime(Long.valueOf(value[1]));
			list.add(e);
		}
		return list;
	}
}
